package org.fuxia.im.core.socket.impl;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import org.apache.log4j.Logger;
import org.fuxia.im.core.entries.ECPacket;
import org.fuxia.im.core.entries.ECResponse;
import org.fuxia.im.core.entries.ECSession;
import org.fuxia.im.core.iconst.ServerConst;
import org.fuxia.im.core.server.ECSessionMananger;
import org.fuxia.im.core.service.DispathService;
import org.fuxia.im.core.socket.ECServer;
import org.fuxia.im.core.utils.ObjectUtil;

/**
 * 
 * @author Fuxia
 * @create @date 2011-12-27-下午9:04:00
 * @comment
 */
public class ECNioServer implements ECServer
{
	private static final Logger	logger			= Logger.getLogger(ECNioServer.class);
	ByteBuffer					packetBuffer	= ByteBuffer.allocate(1024 * 1024);
	private boolean				isCreated		= false;
	private InetSocketAddress	address;
	private ServerSocketChannel	serverChannel;
	private int					port;
	private Map<String, Object>	clients;
	private Selector			selector;
	private ECSessionMananger	sessionManager;

	private void initContext()
	{
		sessionManager = ECSessionMananger.getInstance();
		context.put(ServerConst.SESSION_MANAGER, sessionManager);
		context.put(ServerConst.PORT, this.port);
		context.put(ServerConst.SERVER_CHANNEL, this.serverChannel);
		context.put(ServerConst.SERVER_SELECTOR, this.selector);
		context.put(ServerConst.CLIENTS, this.clients);
	}

	public void bind(int port) throws Exception
	{

		try
		{
			serverChannel = ServerSocketChannel.open();
			address = new InetSocketAddress(port);
			serverChannel.socket().bind(address);
			serverChannel.socket().setReceiveBufferSize(1024 * 1024);
			serverChannel.configureBlocking(false);
			selector = Selector.open();
			serverChannel.register(selector, SelectionKey.OP_ACCEPT);
			clients = new HashMap<String, Object>();
			isCreated = true;
			this.port = port;

			initContext();

			logger.info("Server is listen on port :" + port);
		}
		catch (IOException e)
		{
			throw new Exception("Server created failure!!!!");
		}
	}

	public static void sendPacket(ECPacket packet , SocketChannel client) throws Exception
	{
		try
		{
			byte[] message = ObjectUtil.getByteByJson(packet);
			ByteBuffer buffer = ByteBuffer.allocate(message.length);
			buffer.put(message);
			buffer.flip();
			client.write(buffer);
			logger.info(packet);
		}
		catch (IOException e)
		{
			logger.error("信息发送失败", e);
		}

	}

	public static void writeResponse(ECResponse resp , SocketChannel client) throws Exception
	{
		try
		{
			byte[] result = ObjectUtil.getByteByJson(resp);
			ByteBuffer buffer = ByteBuffer.allocate(result.length);
			buffer.put(result);
			buffer.flip();
			logger.info("响应消息长度：" + result.length);
			client.write(buffer);
		}
		catch (UnsupportedEncodingException e)
		{
			logger.error("错误的字符集", e);
			throw e;
		}
		catch (IOException e)
		{
			//e.printStackTrace();
			logger.error("发送信息失败,需要检查客户端是否处于连接状态", e);
			throw e;
		}
	}

	public void start()
	{
		while (isCreated)
		{
			try
			{
				/**
				 * 这里是阻塞方法。当selector注册后的时间发生的时候。那么代码就会继续执行，否则当前执行就在这里阻塞
				 */
				int keys = selector.select();
				if (keys == 0)
				{
					continue;
				}
				Iterator<SelectionKey> keyIter = selector.selectedKeys().iterator();
				while (keyIter.hasNext())
				{
					SelectionKey key = keyIter.next();
					try
					{
						if (key.isAcceptable())
						{
							ServerSocketChannel server = (ServerSocketChannel) key.channel();
							SocketChannel sc = server.accept();
							sc.configureBlocking(false);
							// 创建未登录用户的session
							ECSession curSession = new ECSession();
							curSession.setAttribute(ServerConst.CHANNEL.name(), sc);
							curSession.setInetAddress(sc.socket().getInetAddress().getHostAddress());
							sessionManager
							.addTempedSession(curSession.getInetAddress() + ":" + sc.hashCode(), curSession);
							//--end--
							sc.register(context.get(ServerConst.SERVER_SELECTOR, Selector.class), SelectionKey.OP_READ);
							logger.info("a client is connected,ip address is [" + curSession.getInetAddress() + "]");
						}
						else if (key.isReadable())
						{
							/**
							 *   总数据包的长度有 50个字节  前面4个字节就是这个f分组的长度。
							 *   从第五个字节
							 * byte[] = byte[1,1,1,1];
							 */
							SocketChannel sc = (SocketChannel) key.channel();
							packetBuffer.clear();
							sc.read(packetBuffer);
							packetBuffer.flip();
							int pSize = packetBuffer.getInt();
							byte[] objs = Arrays.copyOfRange(packetBuffer.array(), packetBuffer.position(), pSize + 4);
							try
							{
								ECPacket packet = ObjectUtil.getObjectFromByte(objs, ECPacket.class);
								if (packet != null)
								{
									packet.addParam(ServerConst.SESSION_ID.name(), sc.socket().getInetAddress()
											.getHostAddress()
											+ ":" + sc.hashCode());
									DispathService.dispath(packet);
								}
							}
							catch (Exception e)
							{
								key.cancel();
								e.printStackTrace();
								//e.printStackTrace();
								logger.error("系统出现错误，" + e.getMessage(), e);
							}
						}
					}
					catch (IOException e)
					{
						SocketChannel channel = (SocketChannel) key.channel();
						String id = channel.socket().getInetAddress().getHostAddress() + ":" + channel.hashCode();
						ECSessionMananger.getInstance().removeTempedSession(id);
						logger.info("当前已经打开的session总数："
								+ ECSessionMananger.getInstance()
										.getSessionCounts());
						logger.error("客户端 {"
								+ channel.socket().getInetAddress()
										.getHostAddress() + "} 异常退出");
						key.cancel();
					}
					keyIter.remove();
				}
			}
			catch (IOException e)
			{
				//e.printStackTrace();
				logger.error(e.getMessage());
			}

		}
	}

	public void shutdown()
	{
		isCreated = false;
	}

	/**
	 * 获取端口号
	 * 
	 * @return
	 */
	public int getPort()
	{
		return this.port;
	}

	/**
	 * 获取连接的客户端数
	 * 
	 * @return
	 */
	public int getClientCount()
	{
		return clients.keySet().size();
	}
}
